package com.flute.haflute.agent.job;

import java.io.IOException;
import java.util.ArrayList;

import org.apache.mina.core.session.IoSession;

import com.flute.haflute.agent.JobControlCenterManager;
import com.flute.haflute.agent.JobSlaveManager;
import com.flute.haflute.agent.topo.ClusterMessageListener;
import com.flute.haflute.common.ClusterConstants;
import com.flute.haflute.common.ClusterConstants.JOB_ROLL;
import com.flute.haflute.net.messaging.Message;
import com.flute.haflute.net.messaging.MessageFactory;
import com.flute.haflute.net.messaging.MessageSessionPair;
import com.flute.haflute.tools.Utils;

/**
 * 每一个节点监控JOB计划分配的入口
 */
public class JobListener4Node implements ClusterMessageListener {

	/** jobRequests JOB请求的缓存 */
	private ArrayList<MessageSessionPair> jobRequests;

	public JobListener4Node() {
		jobRequests = new ArrayList<MessageSessionPair>();
		startHandleThread();
	}

	/*
	 * @see
	 * com.flute.haflute.agent.topo.ClusterMessageListener#handleMessage(com
	 * .flute.haflute.net.messaging.Message,
	 * org.apache.mina.core.session.IoSession)
	 */
	@Override
	public void asyncHandle(Message message, IoSession session) {
		synchronized (jobRequests) {
			jobRequests.add(new MessageSessionPair(message, session));
			jobRequests.notifyAll();
		}
	}

	private void startHandleThread() {
		Thread t = new Thread(new Runnable() {
			@Override
			public void run() {
				while (true) {
					if (jobRequests.size() > 0) {
						MessageSessionPair pair;
						synchronized (jobRequests) {
							pair = jobRequests.remove(0);
						}

						allcoateResource4Task(pair.getMessage(),
								pair.getSession());

					} else {
						try {
							synchronized (jobRequests) {
								if (jobRequests.size() == 0)
									jobRequests.wait();
							}
						} catch (InterruptedException e) {
						}
					}
				}
			}
		});
		t.setName("JobListener4Node Dealer");
		t.start();
	}

	/**
	 * 启动该任务执行需要的子JVM，分配资源等
	 * 
	 * @param message
	 * @param session
	 */
	public void allcoateResource4Task(Message message, IoSession session) {
		
		if (message.getMessageType().equals(Message.MessageType.JOB_NEW.name())) {
			byte[] jobRollBytes = message
					.getHeader(ClusterConstants.JOB_ROLE_TAG);
			int jobRollIndex = Utils.bytes2int(jobRollBytes);
			JOB_ROLL role = ClusterConstants.JOB_ROLL.values()[jobRollIndex];
			if (role == ClusterConstants.JOB_ROLL.MASTER) {
				// 该JOB的控制节点是本节点，启动主控JVM
				new JobControlCenterManager().startControllJvm();
			} else {
				// 本节点是任务执行节点之一，启动相关子JVM
				String master = new String(
						message.getHeader(ClusterConstants.JOB_MASTER_TAG));
				JobSlaveManager.startTaskExecutorJvm(master);
			}

			Message response = MessageFactory.getNewJobResponseMessage(true,
					message.getFrom().getHostAddress());
			try {
				// 发送资源准备完成的消息
				Utils.sendMessage(session, response);
				session.close(false);
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

}
